package ltd.dujiabao.orderservice.service;

import lombok.extern.slf4j.Slf4j;
import ltd.dujiabao.orderservice.dao.OutboxMessageRepository;
import ltd.dujiabao.orderservice.domain.entity.OutboxMessage;
import ltd.dujiabao.orderservice.domain.enums.MessageStatus;
import ltd.dujiabao.orderservice.properties.OutboxProperties;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.List;


/**
 * @author Bowie Du
 * @date 2025/7/9 22:27
 */
@Slf4j
@Component
public class OutboxWorker {

    @Autowired
    private OutboxMessageRepository outboxMessageRepository;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private OutboxProperties outboxProperties;

    @Scheduled(fixedRate = 5000)
    @Transactional(rollbackFor = Exception.class)
    public void process() {
        log.info("Start processing outbox messages");
        List<OutboxMessage> outboxMessages =
                outboxMessageRepository.findForUpdateSkipLocked(MessageStatus.WAITING.getValue(), 100);
        for (OutboxMessage outboxMessage : outboxMessages) {
            sendMessage(outboxMessage);
        }
        outboxMessageRepository.saveAll(outboxMessages);
    }

    private void sendMessage(OutboxMessage outboxMessage) {
        try {
            // 消息id需要传递，用于ack
            CorrelationData correlationData = new CorrelationData(outboxMessage.getMessageId());
            rabbitTemplate.convertAndSend(
                    outboxMessage.getExchange(),
                    outboxMessage.getRoutingKey(),
                    outboxMessage.getMessage(),
                    message -> {
                        message.getMessageProperties().setMessageId(outboxMessage.getMessageId());
                        return message;
                    },
                    correlationData);
            outboxMessage.setStatus(MessageStatus.SUCCESS.getValue())
                    .setUpdatedAt(LocalDateTime.now());
            log.info("send message success. messageId={}", outboxMessage.getMessageId());
        } catch (AmqpException e) {
            log.error("send message error. messageId={}", outboxMessage.getMessageId(), e);
            if (outboxMessage.increaseAndGetRetryCount() == outboxProperties.getMaxRetry()) {
                outboxMessage.setStatus(MessageStatus.FAILED.getValue())
                        .setErrorMsg(e.getMessage())
                        .setUpdatedAt(LocalDateTime.now());
            } else {
                outboxMessage.setStatus(MessageStatus.WAITING.getValue())
                        .setErrorMsg(e.getMessage())
                        .setUpdatedAt(LocalDateTime.now());
            }
        }
    }
}
